#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2019/11/15 下午7:20
# @Author  : yinxin
# @File    : Consumer
# @Software: PyCharm

import inspect
import logging

from kafka import KafkaConsumer

from common import mq_config
from log import hlog

logging.getLogger("kafka").setLevel(logging.WARNING)


class Consumer:
    last_obj = None

    def __init__(self):

        func_name = "Consumer." + inspect.stack()[0][3]
        hlog.enter_func(func_name)

        hlog.info("初始化消息队列连接")

        hlog.var('mq_host', mq_config.host)
        hlog.var('mq_port', mq_config.port)
        hlog.var('mq_group', mq_config.group)
        hlog.var('mq_topic', mq_config.topic)

        self.consumer = KafkaConsumer(mq_config.topic,
                                      group_id=mq_config.group,
                                      bootstrap_servers=[
                                          '{}:{}'.format(mq_config.host, mq_config.port)
                                      ],
                                      max_poll_records=1
                                      )

        hlog.exit_func(func_name)

    @staticmethod
    def get_consumer():

        func_name = "Consumer." + inspect.stack()[0][3]
        hlog.enter_func(func_name)

        if not Consumer.last_obj:
            try:
                Consumer.last_obj = Consumer()
            except BaseException as e:
                hlog.error("消息队列初始化连接异常")
                hlog.error(str(e))
                Consumer.last_obj = None
            else:
                hlog.info("消息队列连接完成")

        hlog.exit_func(func_name)

        return Consumer.last_obj

    def get_msg(self):

        func_name = "Consumer." + inspect.stack()[0][3]
        hlog.enter_func(func_name)

        result = next(self.consumer)

        hlog.var("result", result)

        hlog.exit_func(func_name)

        return result
